package zhanglu.family.socket.textline.syncclient;

import java.net.InetSocketAddress;
import java.nio.charset.Charset;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

import org.apache.mina.core.filterchain.DefaultIoFilterChainBuilder;
import org.apache.mina.core.future.ConnectFuture;
import org.apache.mina.core.service.IoHandlerAdapter;
import org.apache.mina.filter.codec.ProtocolCodecFilter;
import org.apache.mina.filter.codec.textline.TextLineCodecFactory;
import org.apache.mina.transport.socket.nio.NioSocketConnector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import zhanglu.family.socket.bean.ErrMsg;
import zhanglu.family.socket.bean.syncbean.SyncMsgBean;
import zhanglu.family.socket.util.JsonUtil;

public class SyncTcpClient {
    private static Logger         logger   = LoggerFactory.getLogger(SyncTcpClient.class);

    private NioSocketConnector    connector;                                              // 总进程

    private ConnectFuture         connectFuture;                                          // 客户端连接

    private IoHandlerAdapter      handlerAction;                                          // 业务处理类

    private int                   idleTime = 10;                                          // 连接超时时间秒,默认10秒

    private String                ip;                                                     // 服务端IP

    private int                   port;                                                   // 服务端端口

    private String                charset  = "UTF-8";                                     // 编码格式

    private long                  seqId    = 0;                                           // 流水号为0。用于同步消息制定。用户不需要传流水号。发送同步消息的时候。每次都new实例化，所以流水号不会变动，都是0，但因为每次对象不同，所以不会出现消息重叠
    public Map<Long, SyncMsgBean> synMsgMap;                                              // 用于做同步消息处理的map

    /**
     * 实例化对象
     * 默认超时60秒、重连时间间隔5秒、编码UTF-8，可用方法重新设置
     * 
     * @param ip
     *            服务端IP
     * @param port
     *            服务端端口
     */
    public SyncTcpClient(String ip, int port) {
        this.idleTime = 60;
        this.ip = ip;
        this.port = port;
        this.charset = "UTF-8";
        this.synMsgMap = new ConcurrentHashMap<Long, SyncMsgBean>();
    }

    /**
     * 设置超时时间
     * 
     * @param idleTime
     *            超时时间（秒）
     */
    public void setIdleTime(int idleTime) {
        this.idleTime = idleTime;
    }

    /**
     * 设置编码格式
     * 
     * @param charset
     *            编码格式
     */
    public void setCharset(String charset) {
        this.charset = charset;
    }

    /**
     * 启动
     * 
     * @param handlerAction
     *            业务处理类
     */
    public void startClient(IoHandlerAdapter handlerAction) {
        this.handlerAction = handlerAction;
        createClient();
    }

    /**
     * 创建客户端
     */
    private void createClient() {
        this.connector = new NioSocketConnector();
        // 创建接受数据的过滤器
        DefaultIoFilterChainBuilder chain = this.connector.getFilterChain();
        // 设定这个过滤器将一行一行(/r/n)的读取数据
        chain.addLast("myChin", new ProtocolCodecFilter(new TextLineCodecFactory(Charset.forName(this.charset))));// 文本处理

        // 客户端的消息处理器：一个SamplMinaServerHander对象
        this.connector.setHandler(this.handlerAction);
        // set connect timeout
        this.connector.setConnectTimeoutMillis(this.idleTime * 1000);
        this.connectFuture = connector.connect(new InetSocketAddress(this.ip, this.port));
        this.connectFuture.awaitUninterruptibly();// 这一行阻塞，等连接上了继续向下走，否则正在连接的过程中直接发数据会报错
        try {
            this.connectFuture.getSession().write(" ");
            logger.info("客户端连接成功,服务端ip为" + (this.ip + ",服务端端口为" + this.port));
        } catch (Exception e) {
            logger.info("客户端启动失败");
        }
    }

    /**
     * 发送同步消息
     * 
     * @param msg
     *            消息对象
     * @param secondTimeout
     *            超时时间(秒)
     * @return
     */
    public SyncMsgBean send(Object bean, int secondTimeout) {

        SyncMsgBean msgResult = new SyncMsgBean();

        try {

            if (this.connectFuture == null || !connectFuture.isConnected()) {// 没有连接到总线
                msgResult.setErrMsg(ErrMsg.NOCONNECTION);
                return msgResult;
            }

            SyncMsgBean wait = new SyncMsgBean();
            wait.setErrMsg(ErrMsg.WAIT);
            this.synMsgMap.put(seqId, wait);// 这里因为每次发送都会重新new一个对象，所以不会有并发或者消息重叠。key设置为0即可

            this.connectFuture.getSession().write(JsonUtil.ObToJsonStr(bean));

            int interval = 1;// 定时间隔（毫秒）这里影响性能，越高，性能越差
            int tick = 0;// 计数器，定时查询反馈消息
            while (tick <= secondTimeout * 1000) {// 人为阻塞获取消息体
                SyncMsgBean msgReceive = this.synMsgMap.get(seqId);

                if (msgReceive.getErrMsg() == ErrMsg.NONE) {// 如果接受到了数据，则返回
                    msgResult.setJsonMsg(msgReceive.getJsonMsg());
                    msgResult.setErrMsg(ErrMsg.NONE);

                    this.synMsgMap.remove(seqId);

                    break;
                }

                tick++;

                if (tick > secondTimeout * 1000) {// 访问超时，清除同步消息。
                    msgResult.setErrMsg(ErrMsg.CLIENTTIMEOUT);
                    this.synMsgMap.remove(seqId);

                    break;
                }

                Thread.sleep(interval);
            }
        } catch (Exception e) {
            logger.error("连接错误", e);
        }

        return msgResult;
    }

    /**
     * 杀掉整个连接的进程
     */
    public void dispose() {
        if (this.connector != null) {
            this.connector.dispose();
            this.connector = null;
            logger.info("进程已经杀死");
        }
    }

    public void post(Object bean) {
        this.connectFuture.getSession().write(bean);
    }
}
